package com.luboyan.CBillbill.util;

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.CountDownLatch2;
import org.apache.rocketmq.common.message.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;

/**
 * RocketMQ工具类，用于消息的发送，有异步和同步两种方式
 */
public class RocketMQUtil {

    /**
     * 同步发送消息
     *
     * @param producer 消息生产者
     * @param message  发送的消息，配置类中的messageExt继承了message，添加了一些字段
     */
    public static void syncSendMsg(DefaultMQProducer producer, Message message) throws Exception {
        SendResult result = producer.send(message);
        System.out.println(result);
    }

    /**
     * 异步发送消息
     *
     * @param producer 消息生产者
     * @param msg      发送的消息，配置类中的messageExt继承了message，添加了一些字段
     */
    public static void asyncSendMsg(DefaultMQProducer producer, Message msg) throws Exception {
        producer.send(msg, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                Logger logger = LoggerFactory.getLogger(RocketMQUtil.class);
                logger.info("异步发送消息成功，消息id：" + sendResult.getMsgId());
            }

            @Override
            public void onException(Throwable e) {
                e.printStackTrace();
            }
        });
    }


//    public static void asyncSendMsg(DefaultMQProducer producer, Message message) throws Exception{
//        //对消息进行2次发送
//        int messageCount = 2;
//        CountDownLatch2 countDownLatch2 = new CountDownLatch2(messageCount);
//        for (int i = 0; i < messageCount; i++) {
//            producer.send(message, new SendCallback() {
//                @Override
//                public void onSuccess(SendResult sendResult) {
//                    countDownLatch2.countDown();
//                    System.out.println(sendResult.getMsgId());
//                }
//
//                @Override
//                public void onException(Throwable throwable) {
//                    countDownLatch2.countDown();
//                    System.out.println("发生了异常！" + throwable);
//                    throwable.printStackTrace();
//                }
//            });
//        }
//        countDownLatch2.await(5, TimeUnit.SECONDS);
//    }
}
